package com.bungarus.model;

import com.bungarus.busi.Callback;
import com.bungarus.busi.InnerClient;
import com.bungarus.busi.Remedy;
import com.bungarus.busi.ServerLoggingHandler;
import com.bungarus.cluster.*;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelOption;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import org.apache.log4j.Logger;
import org.springframework.data.mongodb.core.MongoTemplate;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CopyOnWriteArrayList;

/**
 * Created by tang on 2018/11/24.
 */
abstract public class AbstractBungarusServer implements Node, BungarusServer, BungarusClient {
    final Logger logger = Logger.getLogger(AbstractBungarusServer.class);

    private NioEventLoopGroup boss = new NioEventLoopGroup();
    private NioEventLoopGroup worker = new NioEventLoopGroup();
    private List<Channel> serverChannels = new ArrayList<>(0);

    protected Map<String, CopyOnWriteArrayList<NodeInfo>> remoteServerNodes =
            new HashMap<>(0);

    private Map<String, InnerClient> clients = new HashMap<>(0);

    private NodeInfo nodeInfo;
    private int status = Node.STATUS_STOP;

    /**
     * the cluster manager MUST ONLY has ONE instance.
     * @return
     */
    abstract public ClusterManager getClusterManager();

    /**
     * channel handler initializer for handling request
     * @return
     */
    abstract public ChannelHandler getChildHandlerInitializer();

    /**
     * mongo template (spring data concept) for storing failed retry messages
     * @return mongo template
     */
    abstract protected MongoTemplate getMongoTemplate4FailedRetry();

    @Override
    public void bind(int[] servicePorts, int[] heartBeatPorts) throws BungarusServerException {
        if(null == getChildHandlerInitializer()) {
            throw new BungarusServerException("server bind error: no handler.");
        }
        if(null == servicePorts) {
            throw new BungarusServerException("server bind error: no service listening port.");
        }

        ServerBootstrap server = new ServerBootstrap();
        server.group(boss, worker)
                .channel(NioServerSocketChannel.class)
                .option(ChannelOption.SO_BACKLOG, 168)
                .handler(new ServerLoggingHandler())
                .childHandler(getChildHandlerInitializer());

        try {
            for(int port : servicePorts){
                serverChannels.add(server.bind(port).sync().channel());
            }
            if(null != heartBeatPorts) {
                for (int port : heartBeatPorts) {
                    serverChannels.add(server.bind(port).sync().channel());
                }
            }
        }
        catch (Exception e){
            worker.shutdownGracefully();
            boss.shutdownGracefully();
            throw new BungarusServerException(e);
        }
        logger.info("Server initialized successfully.");
    }

    @Override
    public void initRemedy() throws Exception {
        Remedy.start(this);
    }

    @Override
    public void initClusterAndConnectRemoteNode() throws BungarusServerException{
        try {
            //register myself into a cluster
            register2Cluster();
            //connect to remotes servers (xx layer, yy layer) as client
            boolean again;
            do {
                getNodesInCluster();
                again = remoteServerNodes.size() != getRemoteNodeTags().length;
                if(again) {
                    logger.info("FAILED TO GET CLUSTER NODES!!! Try again...");
                }
                else {
                    for(String CLUSTER_TAG : getRemoteNodeTags()) {
                        if(remoteServerNodes.get(CLUSTER_TAG).size() == 0){
                            logger.info("NO NODES FOUND IN CLUSTER " + CLUSTER_TAG + ", Try again...");
                            again = true;
                            break;
                        }
                    }
                }
                if(again) {
                    Thread.sleep(2000);
                }
            } while (again);

            for (String CLUSTER_TAG : getRemoteNodeTags()) {
                int index = this.nodeInfo.getNodeId() % remoteServerNodes.get(CLUSTER_TAG).size();
                CopyOnWriteArrayList<NodeInfo> queue = remoteServerNodes.get(CLUSTER_TAG);
                clients.put(CLUSTER_TAG, this.connect(index, queue));
            }
            this.setStatus(Node.STATUS_RUNNING);
        }
        catch (Exception e) {
            throw new BungarusServerException(e);
        }
    }

    @Override
    public void listen(Callback<Object, Void> callback) throws BungarusServerException{
        if(this.serverChannels.size() > 0){
            logger.info("Server start and is listening to ...");
            try {
                for (Channel channel : this.serverChannels) {
                    channel.closeFuture().sync();
                }
            }
            catch (Exception e){
                throw new BungarusServerException(e);
            }
            finally {
                getClusterManager().close();
                if(null != callback) {
                    callback.call(null);
                }
                worker.shutdownGracefully();
                boss.shutdownGracefully();
            }
        }
    }

    @Override
    public void register2Cluster() throws RegisterNodeException {
        this.nodeInfo = getClusterManager().registerNode(this);
    }

    protected void getNodesInCluster() {
        List<ClusterInfo> clusters;
        for (String CLUSTER_TAG : getRemoteNodeTags()) {
            try {
                clusters = getClusterManager().clusters(CLUSTER_TAG);
                if (null != clusters && clusters.size() > 0) {
                    int index = this.nodeInfo.getNodeId() % clusters.size();
                    List<NodeInfo> nodes = getClusterManager().nodes(CLUSTER_TAG, clusters.get(index).getClusterID());
                    remoteServerNodes.put(CLUSTER_TAG, new CopyOnWriteArrayList<>(nodes));
                }
            }
            catch (Exception e) {
                logger.info(e.getMessage());
                e.printStackTrace();
            }
        }
    }

    @Override
    public InnerClient connect(int index, CopyOnWriteArrayList<NodeInfo> queue) throws BungarusClientException {
        InnerClient client = new InnerClient(index, queue, getMongoTemplate4FailedRetry());
        client.run(getTag());
        return client;
    }

    @Override
    public Map<String, InnerClient> getInnerClients() {
        return clients;
    }

    @Override
    public int getStatus() {
        return this.status;
    }

    @Override
    public void setStatus(int status) throws Exception {
        this.nodeInfo.setStatus(status);
        this.status = status;
        getClusterManager().setNodeData(this.nodeInfo);
    }

    @Override
    public InnerClient getInnerClientConnectingToBackendClusterNode(String clusterTag) {
        return clients.get(clusterTag);
    }

    @Override
    public int availableProtocol() {
        return Node.PROTOCOL_TCP;
    }

    @Override
    public void shutdown() throws BungarusServerException {
        logger.info("Server wants to be shutted down but nothing happened. ");
    }

}
